package com.zlm.app;

import com.zlm.bean.OrderEvent;
import com.zlm.bean.OrderResult;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.net.URL;
import java.util.List;
import java.util.Map;

/**
 * @author Harbour
 * @create 2021-05-28 17:56
 */
public class OrderTimeoutWithCEPApp {
    public static void main(String[] args) throws Exception {

        // step 1 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // step 2 获取数据源
        URL resource = OrderTimeoutWithCEPApp.class.getResource("/OrderLog.csv");
        DataStream<String> inputDataStream = env.readTextFile(resource.getPath());

        // step 3 数据清洗、设置时间语义
        SingleOutputStreamOperator<OrderEvent> orderEventStream = inputDataStream.map(new MapFunction<String, OrderEvent>() {
            @Override
            public OrderEvent map(String s) throws Exception {
                String[] fields = s.split(",");
                return new OrderEvent(
                        Long.parseLong(fields[0]),
                        fields[1],
                        fields[2],
                        Long.parseLong(fields[3])
                );
            }
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
            @Override
            public long extractAscendingTimestamp(OrderEvent orderEvent) {
                return orderEvent.getTimestamp() * 1000L;
            }
        });

        // step 4 定义一个匹配时间窗口的 pattern
        Pattern<OrderEvent, OrderEvent> orderPayPattern = Pattern.<OrderEvent>begin("create").where(new SimpleCondition<OrderEvent>() {
            @Override
            public boolean filter(OrderEvent orderEvent) throws Exception {
                return "create".equalsIgnoreCase(orderEvent.getEventType());
            }
        }).followedBy("pay").where(new SimpleCondition<OrderEvent>() {
            @Override
            public boolean filter(OrderEvent orderEvent) throws Exception {
                return "pay".equalsIgnoreCase(orderEvent.getEventType());
            }
        }).within(Time.minutes(15));

        PatternStream<OrderEvent> patternStream = CEP.pattern(orderEventStream.keyBy(OrderEvent::getOrderId), orderPayPattern);
        OutputTag<OrderResult> orderTimeoutOutputTag =
                new OutputTag<OrderResult>("orderTimeOut") {};

        // step 5 处理逻辑
        SingleOutputStreamOperator<OrderResult> resultStream = patternStream.select(
                orderTimeoutOutputTag,
                new OrderTimeoutFunction(),
                new OrderSelectFunction()
        );

        resultStream.print("payed");
        resultStream.getSideOutput(orderTimeoutOutputTag).print("timeout");

        env.execute("OrderTimeoutDetectApp");

    }

    private static class OrderTimeoutFunction implements PatternTimeoutFunction<OrderEvent, OrderResult> {
        @Override
        public OrderResult timeout(Map<String, List<OrderEvent>> map, long l) throws Exception {
            Long timeoutOrderId = map.get("create").iterator().next().getOrderId();
            return new OrderResult(timeoutOrderId, "timeout");
        }
    }

    private static class OrderSelectFunction implements PatternSelectFunction<OrderEvent, OrderResult> {
        @Override
        public OrderResult select(Map<String, List<OrderEvent>> map) throws Exception {
            return new OrderResult(
                    map.get("pay").iterator().next().getOrderId(),
                    "payed"
            );
        }
    }
}
